package org.example.task;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.*;

/**
 * 消息发送任务
 */
@Component
public class MqTask implements CommandLineRunner {

    @Autowired
    public RocketMQTemplate rocketMQTemplate;

    @Override
    public void run(String... args) throws Exception {
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            try {
                Thread.sleep(1000 * 10);
                int activeThreadCount = Thread.activeCount();
                // 消息内容
                String msg = String.format("当前日期和时间【%s】\n发送消息【今天天气不错】\n当前线程数【%s】"
                        , new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())
                        , activeThreadCount);
                String topicAndTap = "msg:user"; // 指定主题和分区
                String topic = "msg"; // 指定主题和默认分区
                // 消息异步发送到指定主题分区
                rocketMQTemplate.asyncSend(topicAndTap, msg, new SendCallback() {
                    // 发送成功的回调方法
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        String brokerName = sendResult.getMessageQueue().getBrokerName();
                        String msgId = sendResult.getMsgId();
                        String status = sendResult.getSendStatus().name();
                        System.out.printf("消息发送成功: \nbrokerName【%s】\nmsgId【%s】\nstatus【%s】%n"
                                , brokerName, msgId, status);
                    }

                    // 发送失败的回调方法
                    @Override
                    public void onException(Throwable throwable) {
                        throwable.printStackTrace();
                        System.out.printf("消息发送出现错误: 【%s】\n消息内容: 【%s】%n"
                                , throwable.getMessage(), msg);
                    }

                });
            } catch (Exception e) {
                e.printStackTrace();
                System.out.printf("消息发送出现错误: 【%s】%n", e.getMessage());
            }
        }, 0, 10, TimeUnit.SECONDS);
    }

}
